package github.sf.fw.tools.db.impl.doris;


import github.sf.fw.tools.db.DSProperty;
import github.sf.fw.tools.db.impl.mysql.MysqlDataWriter;
import github.sf.fw.tools.db.model.ColumnInfo;
import github.sf.fw.tools.db.model.TableInfo;
import github.sf.fw.utils.StrUtil;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class DorisDataWriter extends MysqlDataWriter {
    public DorisDataWriter(DSProperty dsProperty) {
        super(dsProperty);
    }

    @Override
    public String createDDL(TableInfo tableInfo, String tableName) {
        List<ColumnInfo> columnInfoList = tableInfo.getColumnInfoList();
        if (CollectionUtils.isEmpty(columnInfoList)) {
            throw new IllegalArgumentException("列信息为空，不能生成表创建语句:" + tableName);
        }
        //doris的建表语句
//        StringBuilder sb = new StringBuilder("DROP TABLE IF EXISTS `" + tableName + "`;\ncreate table ");
        StringBuilder sb = new StringBuilder("CREATE TABLE IF NOT EXISTS `" + tableName + "` ");
//        sb.append(tableName);
        sb.append("(\n");
        List<String> primaryKeys = new ArrayList<>();
        for (ColumnInfo columnInfo : columnInfoList) {
            String columnType = pureColumnType(columnInfo.getType());
            sb.append("`")
                    .append(columnInfo.getName())
                    .append("`")
                    .append(" ")
                    .append(columnType);
            if ("PRI".equals(columnInfo.getColumnKey())) {
                primaryKeys.add(columnInfo.getName());
            }
            sb.append(",\n");
        }
        sb.deleteCharAt(sb.length() - 2);
        sb.append(")");
        if (!primaryKeys.isEmpty()) {
            String uniqueKeys = primaryKeys.stream().map(k -> "`" + k + "`").collect(Collectors.joining(",", "(", ")"));
            sb.append(" UNIQUE KEY").append(uniqueKeys);

        }
        String distributionColumns = columnInfoList.stream()
                .filter(columnInfo -> !StrUtil.isEmpty(columnInfo.getType()) && columnInfo.getType().toLowerCase().contains("int"))
                .map(ColumnInfo::getName)
                .map(k -> "`" + k + "`")
                .collect(Collectors.joining(",", "(", ")"));
        sb.append(" DISTRIBUTED BY HASH").append(distributionColumns).append(" BUCKETS 1");
        sb.append(" PROPERTIES (\"replication_allocation\" = \"tag.location.default: 1\");");
        return sb.toString();
    }

    @Override
    public boolean createTable(TableInfo tableInfo) {
        String ddl = createDDL(tableInfo, tableInfo.getTableExtraInfo().getTargetTableName());
        return executeDDL(ddl);
    }

    private String pureColumnType(String columnType) {
        if (StrUtil.isEmpty(columnType)) {
            return columnType;
        }
        return columnType.split(" ")[0];
    }
}
